package utils

import (
	"context"
	"fmt"
	"log"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"

	amqp "github.com/rabbitmq/amqp091-go"
)

/*
 +----------------------------------------------------------------------
 + Title        : rabbit消息队列【单例模式支持并发安全】
 + Author       : huwl
 + Version      : 1.0.0
 + Initial-Time : 2021-09-22
 + Last-time    : 这个文件最后修改的时间 + 修改人的名称
 + Desc         : rabbit
 +----------------------------------------------------------------------
*/

var (
	rabbitInstance *RabbitInstance
	rabbitMu       sync.RWMutex
	rabbitInitErr  error
)

// 定义Rabbit结构体
type RabbitInstance struct {
	connect *amqp.Connection
	Ch      *amqp.Channel
}

// Rabbit配置结构体
type RabbitConf struct {
	Ip       string `yaml:"ip" json:"ip"`
	Port     string `yaml:"port" json:"port"`
	Vhost    string `yaml:"vhost" json:"vhost"`
	Username string `yaml:"username" json:"username"`
	Password string `yaml:"password" json:"password"`
}

// 定义交换机参数结构体
type ExchangeDeclareParams struct {
	// 是否创建交换机
	IsExchangeDeclare bool
	// 名称
	Name string
	// 种类
	Kind string
	// 持久的
	Durable bool
	// 自动删除
	AutoDelete bool
	// 间隔
	Internal bool
	// 不等待、同步
	NoWait bool
	// 额外参数
	AmqpTable amqp.Table
}

// 定义队列参数结构体
type QueueDeclareParams struct {
	// 是否创建队列
	IsQueueDeclare bool
	// 队列
	Queue string
	// 持久的
	Durable bool
	// 自动删除
	AutoDelete bool
	// 专用的
	Exclusive bool
	// 不等待、同步
	NoWait    bool
	AmqpTable amqp.Table

	// 绑定交换机参数
	QueueBindParam QueueBindParam
}

// 绑定交换机参数结构体
type QueueBindParam struct {
	QueueName string
	// 路由key
	RoutingKey   string
	ExchangeName string
	NoWait       bool
	AmqpTable    map[string]interface{}
}

// 定义发布消息结构体
type PublishWithCtx struct {
	// 交换机
	Exchange string
	// 路由key
	RoutingKey string
	// 强制性的
	Mandatory bool
	// 立刻的
	Immediate bool
	// 消息包
	AmqpPublishing amqp.Publishing
}

// 定义消费消息参数结构体
type ConsumerMessageParam struct {
	// 队列
	Queue string
	// 消费者
	Consumer string
	// 是否自动回复
	AutoAck bool
	// 是否专用的
	Exclusive bool
	NoLocal   bool
	NoWait    bool
	ArgsTable amqp.Table
	// 是否重试
	IsRetry bool
	// 重试次数
	RetryCount int
	// 是否拒绝
	IsReject bool

	// 不确认
	Nack bool
	// 批量不确认
	NackMultiple bool
	// 是否重新入队,进入死信队列
	NackRequeue bool
}

/**
 * 创建rabbit实例
 *
 * @param bool isDialConfigMode --------------------------必传
 * @return *RabbitInstance, error
 * @author huwl
 */
func NewRabbitInstance(isDialConfigMode ...bool) (*RabbitInstance, error) {
	rabbitConf := &RabbitConf{}
	// +++++++++++++++++ 使用本地配置方式开始 +++++++++++++++++
	// cfg, err := GetConfig("rabbitmq", "yml")
	// if err != nil {
	// 	return nil, err
	// }

	// rabbitMQConf.Ip = cfg.GetString(Env + ".ip")
	// rabbitMQConf.Port = cfg.GetString(Env + ".port")
	// rabbitMQConf.Vhost = cfg.GetString(Env + ".vhost")
	// rabbitMQConf.Username = cfg.GetString(Env + ".username")
	// rabbitMQConf.Password = cfg.GetString(Env + ".password")
	// +++++++++++++++++ 使用本地配置方式结束 +++++++++++++++++

	// +++++++++++++++++ 使用配置中心方式开始 +++++++++++++++++
	InitConfigCenter(RabbitConfigCenter, rabbitConf)
	// +++++++++++++++++ 使用配置中心方式结束 +++++++++++++++++

	url := "amqp://" + rabbitConf.Username + ":" + rabbitConf.Password + "@" + rabbitConf.Ip + ":" + rabbitConf.Port + "/" + rabbitConf.Vhost

	isDialConfig := false
	if len(isDialConfigMode) > 0 {
		isDialConfig = isDialConfigMode[0]
	}

	var conn *amqp.Connection
	var err error
	if isDialConfig {
		config := amqp.Config{
			Vhost:      rabbitConf.Vhost,
			Properties: amqp.NewConnectionProperties(),
		}

		config.Properties.SetClientConnectionName("producer-with-confirms")

		conn, err = amqp.DialConfig(url, config)
	} else {
		conn, err = amqp.Dial(url)
	}
	if err != nil {
		return nil, err
	}

	// 创建通道
	ch, err := conn.Channel()
	if err != nil {
		return nil, err
	}

	return &RabbitInstance{
		connect: conn,
		Ch:      ch,
	}, nil
}

/**
 * 获取单列模式：rabbit实例
 *
 * @author huwl
 * @date 2025/11/07 12:19:48
 */
func GetSingleRabbitInstance() (*RabbitInstance, error) {
	// 使用双重检查锁定
	rabbitMu.RLock()
	if rabbitInstance != nil {
		defer rabbitMu.RUnlock()
		return rabbitInstance, rabbitInitErr
	}
	rabbitMu.RUnlock()

	rabbitMu.Lock()
	defer rabbitMu.Unlock()

	if rabbitInstance != nil {
		return rabbitInstance, rabbitInitErr
	}

	rabbitInstance, rabbitInitErr = NewRabbitInstance()
	return rabbitInstance, rabbitInitErr
}

/**
 * 申请交换机
 *
 * @param ExchangeDeclareParams exchangeDeclareParams必传
 * @return error
 * @author huwl
 */
func (r *RabbitInstance) ExchangeDeclare(exchangeDeclareParams ExchangeDeclareParams) error {
	return r.Ch.ExchangeDeclare(
		// 名称
		exchangeDeclareParams.Name,
		// 种类
		exchangeDeclareParams.Kind,
		// 持久的
		exchangeDeclareParams.Durable,
		// 自动删除,当没有队列与其绑定的时候如果设置为true则会自动删除该交换机
		exchangeDeclareParams.AutoDelete,
		// 是否为内部，true表示是。客户端无法直接发送msg到内部交换器，只有交换器可以发送msg到内部交换器。
		exchangeDeclareParams.Internal,
		// 阻塞：表示创建交换器的请求发送后，阻塞等待RMQ Server返回信息。
		// 非阻塞：不会阻塞等待RMQ # Server的返回信息，而RMQ Server也不会返回信息。（不推荐使用）
		exchangeDeclareParams.NoWait,
		// 额外参数
		exchangeDeclareParams.AmqpTable,
	)
}

/**
 * 申请队列
 *
 * @param []QueueDeclareParams queueDeclareParams ---必传
 * @return error
 * @author huwl
 */
func (r *RabbitInstance) QueueDeclare(queueDeclareParams []QueueDeclareParams) error {
	for _, queueDeclareParam := range queueDeclareParams {
		// 创建队列
		_, err := r.Ch.QueueDeclare(
			queueDeclareParam.Queue,      // Queuename
			queueDeclareParam.Durable,    // durable：持久的
			queueDeclareParam.AutoDelete, // autoDelete：自动删除
			queueDeclareParam.Exclusive,  // exclusive：专用的
			queueDeclareParam.NoWait,     // noWait：不等待、同步
			queueDeclareParam.AmqpTable,
			// amqp.Table{
			// 	// 设置队列最大长度
			// 	"x-max-length": 20,
			// 	// 发布的消息在队列中存在多长时间后被取消（单位毫秒)
			// 	"x-message-ttl": 20000,
			// 	// 当Queue（队列）在指定的时间未被访问，则队列将被自动删除
			// 	"x-expires": 15000,
			// 	// 设置队列溢出行为,当达到队列的最大长度时,消息会发生什么,有效值为 Drop Head 或 Reject Publish;
			// 	"x-overflow": "Drop Head",
			// 	// 设置只有一个消费者
			// 	"x-single-active-consumer": true,
			// 	// 设置队列最大容量
			// 	"x-max-length-bytes": 1000,
			// 	// 队列支持的最大优先级数目;如果没有设置，队列将不支持消息优先级,插队的意思
			// 	"x-max-priority": 7,
			// // 惰性队列
			// "x-queue-mode": "lazy",
			// },
		)
		if err != nil {
			return err
		}
	}

	return nil
}

/**
 * 交换机绑定队列
 *
 * @param QueueBindParam queueBindParam 必传
 * @return error
 * @author huwl
 */
func (r *RabbitInstance) ExchangeBindQueue(queueBindParam QueueBindParam) error {
	return r.Ch.QueueBind(
		queueBindParam.QueueName,
		queueBindParam.RoutingKey,
		queueBindParam.ExchangeName,
		queueBindParam.NoWait,
		queueBindParam.AmqpTable,
	)
}

/**
 * 发布消息
 *
 * @param PublishWithCtx publishWithCtx 必传
 * @return error
 * @author huwl
 */
func (r *RabbitInstance) PublishMessage(publishWithCtx PublishWithCtx) error {
	return r.Ch.PublishWithContext(ctx,
		publishWithCtx.Exchange,       // exchange：交换机
		publishWithCtx.RoutingKey,     // routing key：路由名
		publishWithCtx.Mandatory,      // mandatory：强制性的
		publishWithCtx.Immediate,      // immediate：立刻的
		publishWithCtx.AmqpPublishing, // 消息包
		// // 消息包
		// amqp.Publishing{
		// 	ContentType: msgType,
		// 	Body:        message,
		// 	// 设置消息持久化
		// 	DeliveryMode: 2,
		// }
	)
}

/**
 * 基础生产者【包含: 申请交换机、申请队列、交换机绑定队列、发布消息】
 *
 * @param ExchangeDeclareParams exchangeDeclareParams必传
 * @param QueueDeclareParams queueDeclareParams -----必传
 * @param PublishWithCtx publishWithCtx 必传
 * @return error
 * @author huwl
 */
func (r *RabbitInstance) BaseProducer(exchangeDeclareParams ExchangeDeclareParams, queueDeclareParams []QueueDeclareParams, publishWithCtx PublishWithCtx, isQueueBind ...bool) error {
	defer r.Ch.Close()

	// 是否创建交换机
	if exchangeDeclareParams.IsExchangeDeclare {
		err := r.Ch.ExchangeDeclare(
			// 名称
			exchangeDeclareParams.Name,
			// 种类
			exchangeDeclareParams.Kind,
			// 持久的
			exchangeDeclareParams.Durable,
			// 自动删除
			exchangeDeclareParams.AutoDelete,
			// 是否为内部，true表示是。客户端无法直接发送msg到内部交换器，只有交换器可以发送msg到内部交换器。
			exchangeDeclareParams.Internal,
			// 阻塞：表示创建交换器的请求发送后，阻塞等待RMQ Server返回信息。
			// 非阻塞：不会阻塞等待RMQ # Server的返回信息，而RMQ Server也不会返回信息。（不推荐使用）
			exchangeDeclareParams.NoWait,
			// 额外参数
			exchangeDeclareParams.AmqpTable,
		)
		if err != nil {
			return err
		}
	}

	for _, queueDeclareParam := range queueDeclareParams {
		// 是否创建队列
		if queueDeclareParam.IsQueueDeclare {
			_, err := r.Ch.QueueDeclare(
				queueDeclareParam.Queue,      // Queuename
				queueDeclareParam.Durable,    // durable：持久的
				queueDeclareParam.AutoDelete, // autoDelete：自动删除
				queueDeclareParam.Exclusive,  // exclusive：专用的
				queueDeclareParam.NoWait,     // noWait：不等待、同步
				queueDeclareParam.AmqpTable,
				// amqp.Table{
				// 	// 设置队列最大长度
				// 	"x-max-length": 20,
				// 	// 发布的消息在队列中存在多长时间后被取消（单位毫秒)
				// 	"x-message-ttl": 20000,
				// 	// 当Queue（队列）在指定的时间未被访问，则队列将被自动删除
				// 	"x-expires": 15000,
				// 	// 设置队列溢出行为,当达到队列的最大长度时,消息会发生什么,有效值为 Drop Head 或 Reject Publish;
				// 	"x-overflow": "Drop Head",
				// 	// 设置只有一个消费者
				// 	"x-single-active-consumer": true,
				// 	// 设置队列最大容量
				// 	"x-max-length-bytes": 1000,
				// 	// 队列支持的最大优先级数目;如果没有设置，队列将不支持消息优先级,插队的意思
				// 	"x-max-priority": 7,
				// },
			)
			if err != nil {
				return err
			}
		}
	}

	// 交换机绑定队列
	isExchangeBindQueue := false
	if len(isQueueBind) > 0 {
		isExchangeBindQueue = isQueueBind[0]
	}
	if isExchangeBindQueue {
		for _, queue := range queueDeclareParams {
			r.Ch.QueueBind(
				queue.QueueBindParam.QueueName,
				queue.QueueBindParam.RoutingKey,
				queue.QueueBindParam.ExchangeName,
				queue.QueueBindParam.NoWait,
				queue.QueueBindParam.AmqpTable,
			)
		}
	}

	// 设置超时时间
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	// 发布消息
	return r.Ch.PublishWithContext(ctx,
		publishWithCtx.Exchange,       // exchange：交换机
		publishWithCtx.RoutingKey,     // routing key：路由名
		publishWithCtx.Mandatory,      // mandatory：强制性的
		publishWithCtx.Immediate,      // immediate：立刻的
		publishWithCtx.AmqpPublishing, // 消息包
		// // 消息包
		// amqp.Publishing{
		// 	ContentType: msgType,
		// 	Body:        message,
		// 	// 设置消息持久化
		// 	DeliveryMode: 2,
		// }
	)
}

/**
 * 发布可靠性消息
 *
 * @param QueueDeclareParams queueDeclareParams -----必传
 * @param <-chan struct{} publishOkCh --必传
 * @param chan<- *amqp.DeferredConfirmation confirmsCh必传
 * @param <-chan struct{} confirmsDoneCh ------------必传
 * @param chan struct{} exitCh ---------必传
 * @author huwl
 */
func (r *RabbitInstance) PublishReliabilityMessage(publishWithCtx PublishWithCtx, publishOkCh <-chan struct{}, confirmsCh chan<- *amqp.DeferredConfirmation, confirmsDoneCh <-chan struct{}, exitCh chan struct{}) {
	defer r.Ch.Close()

	continuous := false

	// 设置管道确认模式
	if err := r.Ch.Confirm(false); err != nil {
		log.Fatalf("Failed to ch Confirm: %v", err)
	}

	for {
		canPublish := false
		for {
			select {
			case <-confirmsDoneCh:
				log.Println("producer: stopping, all confirms seen")
				return
			case <-publishOkCh:
				log.Println("producer: got the OK to publish")
				canPublish = true
			case <-time.After(time.Second):
				log.Println("producer: still waiting on the OK to publish...")
				continue
			}

			if canPublish {
				break
			}
		}

		// 设置超时时间
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()

		dConfirmation, err := r.Ch.PublishWithDeferredConfirmWithContext(ctx,
			publishWithCtx.Exchange,   // exchange：交换机
			publishWithCtx.RoutingKey, // routing key：路由名
			publishWithCtx.Mandatory,  // mandatory：强制性的
			publishWithCtx.Immediate,  // immediate：立刻的
			// 消息包
			publishWithCtx.AmqpPublishing,
		// amqp.Publishing{
		// 	ContentType: msgType,
		// 	Body:        msg,
		// 	// 设置消息持久化
		// 	DeliveryMode: amqp.Persistent,
		// 	AppId:        "sequential-producer",
		// }
		)

		if err != nil {
			log.Println("发送消息错误: ", err)
		}

		select {
		case <-confirmsDoneCh:
			log.Println("producer: stopping, all confirms seen")
			return
		case confirmsCh <- dConfirmation:
			log.Println("producer: delivered deferred confirm to handler")
			break
		}

		select {
		case <-confirmsDoneCh:
			log.Println("producer: stopping, all confirms seen")
			return
		case <-time.After(time.Millisecond * 250):
			if continuous {
				continue
			} else {
				log.Println("producer: initiating stop")
				close(exitCh)
				select {
				case <-confirmsDoneCh:
					log.Println("producer: stopping, all confirms seen")
					return
				case <-time.After(time.Second * 10):
					log.Println("producer: may be stopping with outstanding confirmations")
					return
				}
			}
		}
	}
}

/**
 * 消息确认
 *
 * @param <-chan amqp.Confirmation confirms ---------必传
 * @author huwl
 */
func (r *RabbitInstance) ConfirmHandler(confirms <-chan amqp.Confirmation) {
	// 如果confirmed.Ack为true，则表示消息已经成功发送和确认；
	// 否则，表示消息发送失败。
	// 在实际应用中，我们可以根据需要选择合适的方式来保证消息的可靠性
	if confirmed := <-confirms; confirmed.Ack {
		fmt.Println("confirmed success", confirmed)
	} else {
		// 当发送失败的时候可以重新发送消息
		fmt.Println("confirm fail", confirmed)
	}
}

/**
 * 发布消息
 *
 * @param <-chan amqp.Return returns ---必传
 * @author huwl
 */
func (r *RabbitInstance) ReturnsHandler(returns <-chan amqp.Return) {
	log.Println("---------------- ReturnsHandler ---")
	// 如果confirmed.Ack为true，则表示消息已经成功发送和确认；
	// 否则，表示消息发送失败。
	// 在实际应用中，我们可以根据需要选择合适的方式来保证消息的可靠性
	// fmt.Println("returns =", returns)
	select {
	case returned := <-returns:
		fmt.Println("returned", returned.ReplyCode)
		fmt.Println("returned", returned.ReplyText)
		if returned.ReplyCode == amqp.NoRoute {
			fmt.Println("路由失败：", returned.ReplyText)
		}
	}
	// for returned := range returns {
	// 	fmt.Println("returned", returned)
	// }
	// if returned, ok := <-returns; ok {
	// 	fmt.Println("returned success", returned)
	// } else {
	// 	// 当发送失败的时候可以重新发送消息
	// 	fmt.Println("returned fail", returned)
	// }
}

/**
 * 开始处理确认
 *
 * @param chan<- struct{} publishOkCh --必传
 * @param <-chan *amqp.DeferredConfirmation confirmsCh必传
 * @param chan struct{} confirmsDoneCh -必传
 * @param <-chan struct{} exitCh -------必传
 * @author huwl
 */
func (r *RabbitInstance) StartConfirmHandler(publishOkCh chan<- struct{}, confirmsCh <-chan *amqp.DeferredConfirmation, confirmsDoneCh chan struct{}, exitCh <-chan struct{}) {
	go func() {
		confirms := make(map[uint64]*amqp.DeferredConfirmation)

		for {
			select {
			case <-exitCh:
				ExitConfirmHandler(confirms, confirmsDoneCh)
				return
			default:
				break
			}

			outstandingConfirmationCount := len(confirms)

			// Note: 8 is arbitrary, you may wish to allow more outstanding confirms before blocking publish
			if outstandingConfirmationCount <= 8 {
				select {
				case publishOkCh <- struct{}{}:
					fmt.Println("confirm handler: sent OK to publish")
				case <-time.After(time.Second * 5):
					fmt.Println("confirm handler: timeout indicating OK to publish (this should never happen!)")
				}
			} else {
				fmt.Println("confirm handler: waiting on outstanding confirmations, blocking publish", outstandingConfirmationCount)
			}

			select {
			case confirmation := <-confirmsCh:
				dtag := confirmation.DeliveryTag
				confirms[dtag] = confirmation
			case <-exitCh:
				ExitConfirmHandler(confirms, confirmsDoneCh)
				return
			}

			CheckConfirmations(confirms)
		}
	}()
}

/**
 * 处理关闭
 *
 * @param chan struct{} exitCh ---------必传
 * @author huwl
 */
func (r *RabbitInstance) SetupCloseHandler(exitCh chan struct{}) {
	c := make(chan os.Signal, 2)
	signal.Notify(c, os.Interrupt, syscall.SIGTERM)
	go func() {
		<-c
		fmt.Println("close handler: Ctrl+C pressed in Terminal")
		close(exitCh)
	}()
}

/**
 * 消费消息
 *
 * @param ConsumerMessageParam consumerMessageParam -必传
 * @author huwl
 */
func (r *RabbitInstance) ConsumeMessage(consumerMessageParam ConsumerMessageParam) {
	messages, err := r.Ch.Consume(
		consumerMessageParam.Queue,
		consumerMessageParam.Consumer,
		consumerMessageParam.AutoAck,
		consumerMessageParam.Exclusive,
		consumerMessageParam.NoLocal,
		consumerMessageParam.NoWait,
		consumerMessageParam.ArgsTable,
	)
	if err != nil {
		log.Fatalf("Failed to message Consume: %v", err)
	}

	var exitChan chan int

	go func() {
		i := 1
		for d := range messages {
			log.Println(consumerMessageParam.Consumer+" "+GetDate()+": received a message = .......", string(d.Body))
			//multiple为true的时候：此次交付和之前没有确认的交付都会在通过同一个通道交付，这在批量处理的时候很有用
			//为false的时候只交付本次。只有该方法执行了，RabbitMQ收到该确认才会将消息删除
			time.Sleep(time.Microsecond * 1)

			// 是否重试
			if consumerMessageParam.IsRetry {
				if i <= consumerMessageParam.RetryCount {
					log.Println("------------- 重试 ")
					// multiple: 是否并发
					// requeue: true(请求服务器将此消息传递给另一个服务器消费者)
					// requeue: false(则消息将被删除丢弃或交付到服务器配置的死信队列。)
					// 不确认，重新入队
					d.Nack(false, true)
				} else {
					// 超过三次不在重试
					log.Println("------------- 3次了，不试了，扔死信队列了 ")
					// r.Ch.Nack(d.DeliveryTag, false, false)
					// 不确认，不重新入队，进入死信队列
					d.Nack(false, false)
				}

				i++
			} else {
				if !consumerMessageParam.AutoAck {
					log.Println("------------- 手动确认 ")

					// 是否拒绝消息
					if consumerMessageParam.IsReject {
						log.Println("------------- 拒绝消息 ")
						// 拒绝消息,进入死信队列
						d.Reject(false)
					} else if consumerMessageParam.Nack { // 不确认
						log.Println("------------- 不确认 ")
						// consumerMessageParam.NackRequeue: true,不确认重新入队
						// consumerMessageParam.NackRequeue: false,不确认不重新入队进入死信队列
						d.Nack(consumerMessageParam.NackMultiple, consumerMessageParam.NackRequeue)
					} else {
						// multiple: true的时候会批量应答
						// multiple: false的时候只应答本次。只有该方法执行了，RabbitMQ收到该确认才会将消息删除
						// 应答方式一
						d.Ack(false)

						// 应答方式二
						// r.Ch.Ack(d.DeliveryTag, false)
					}
				} else {
					log.Println("------------- 自动确认 ")
				}
			}
		}
	}()

	log.Println(" [*] Waiting for messages. To exit press CTRL+C")
	<-exitChan
}

/**
 * 关闭
 * @author huwl
 * @date 2025/11/04 13:41:01
 */
func CloseRabbit() {
	rabbitMu.Lock()
	defer rabbitMu.Unlock()

	if rabbitInstance != nil && rabbitInstance.connect != nil {
		rabbitInstance.Ch.Close()
		rabbitInstance.connect.Close()
		rabbitInstance = nil
		rabbitInitErr = nil
	}
}
